import org.apache.flink.annotation.Public;
import org.apache.flink.streaming.api.functions.source.FromIteratorFunction;

import java.io.Serializable;
import java.util.Iterator;

/**
 * 订单源
 *
 * @author <achao1441470436@gmail.com>
 * @since 2021/11/4 0004 23:43
 */
@Public
public class OrderSource extends FromIteratorFunction<Order> {
    private static final long serialVersionUID = -3541830323543628337L;

    public OrderSource() {
        super(new RateLimitedIterator<>(Customers.unbounded()));
    }


    private static class RateLimitedIterator<T> implements Iterator<T>, Serializable {

        private static final long serialVersionUID = 1L;

        private final Iterator<T> inner;

        private RateLimitedIterator(Iterator<T> inner) {
            this.inner = inner;
        }

        @Override
        public boolean hasNext() {
            return inner.hasNext();
        }

        @Override
        public T next() {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return inner.next();
        }
    }
}
